package SparkExercises.kejian5.lianxi1;

import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import scala.Serializable;

import java.util.List;

public class LianXi1 {
    public static void main(String[] args) {
        /**
         * 练习：有一份新能源车实时上报的车辆动态数据
         * （字段说明：vin：车架号，是车辆的唯一表示，
         * collect_time：采集时间，是数据的上传时间，
         * speed：车速，是当前采集时间所对应的车辆速度），
         * 请写一个spark程序，统计出每辆车每天的最大车速，并将信息输出到控制台，如下图所示：
         */
        SparkConf sparkConf = new SparkConf().setAppName("LianXi1").setMaster("local");
        JavaSparkContext sc = new JavaSparkContext(sparkConf);

        Gson gson1 = new GsonBuilder().create();

        JavaRDD<String> rdd1 = sc.textFile("./data/can_data.json");

        JavaRDD<Car> rdd2 = rdd1.map(new Function<String, Car>() {
            @Override
            public Car call(String s) throws Exception {
                Car car = new Gson().fromJson(s, Car.class);
                Car car1 = new GsonBuilder().create().fromJson(s, Car.class);
                return car1;
            }
        });

        Function<Car, Integer> function = new Function<Car, Integer>() {
            @Override
            public Integer call(Car car) throws Exception {
                return car.getSpeed();
            }
        };

        JavaRDD<Car> carJavaRDD = rdd2.sortBy(function, false, rdd1.getNumPartitions());

        List<Car> collect = carJavaRDD.collect();

        for (Car c :collect){
            System.out.println(c);
        }


    }
}
class Car implements Serializable {
    private String collectTime;
    private Integer speed;
    private String vin;

    public String getCollectTime() {
        return collectTime;
    }

    public void setCollectTime(String collectTime) {
        this.collectTime = collectTime;
    }

    public Integer getSpeed() {
        return speed;
    }

    public void setSpeed(Integer speed) {
        this.speed = speed;
    }

    public String getVin() {
        return vin;
    }

    public void setVin(String vin) {
        this.vin = vin;
    }

    @Override
    public String toString() {
        return "Car{" +
                "collectTime='" + collectTime + '\'' +
                ", speed=" + speed +
                ", vin='" + vin + '\'' +
                '}';
    }
}